package tsif.tcluster.service.proxy.impl;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

import tsif.tcluster.session.TClusterSession;

/**
 * 回调管理器<br>
 * 
 */
public class TClusterCallbacks {
	protected AtomicInteger callbackCounter; // 回调自增Id
	protected ConcurrentMap<Integer, CallBackNode> callbacks; // 回调表
	protected TClusterRpcServiceProxy serviceProxy;

	public TClusterCallbacks(TClusterRpcServiceProxy serviceProxy) {
		callbackCounter = new AtomicInteger();
		callbacks = new ConcurrentHashMap<>();
		this.serviceProxy = serviceProxy;
	}

	/** 添加返回函数, 返回回调Id **/
	public <T> int addCallBack(CallBack<T> callback) {
		// set callback
		int callbackId = 0;
		if (callback != null) {
			callbackId = callbackCounter.incrementAndGet();
			// create node
			CallBackNode node = new CallBackNode();
			node.callbackId = callbackId;
			node.callback = callback;
			node.startTime = System.currentTimeMillis();
			callbacks.put(callbackId, node);
		}
		return callbackId;
	}

	/** 移除回调 **/
	public CallBackNode removeCallBack(int callbackId) {
		return callbacks.remove(callbackId);
	}

	/** 移除回调 **/
	public boolean removeCallBack(CallBackNode node) {
		return callbacks.remove(node.callbackId, node);
	}

	/** 更新回调, 处理超时回调. **/
	public void update() {
		// 检测回调
		if (callbacks.isEmpty()) {
			return;
		}
		long nowTime = System.currentTimeMillis();
		// 遍历检测
		List<CallBackNode> removes = new ArrayList<>();
		for (CallBackNode callbackNode : callbacks.values()) {
			long dt = nowTime - callbackNode.startTime;
			if (dt < callbackNode.callback.getMaxTimeOut()) {
				continue;
			}
			removes.add(callbackNode);
		}

		// call remove
		callBackTimeOut(removes);
		removes.clear();
	}

	// /** 反射调用函数 **/
	// protected static final Method onCallBackMethod;
	// protected static final Method onTimeOutMethod;
	// static {
	// try {
	// onCallBackMethod = CallBack.class.getDeclaredMethod("onCallBack", new Class<?>[] { int.class, String.class, Object.class });
	// onCallBackMethod.setAccessible(true);
	// onTimeOutMethod = CallBack.class.getDeclaredMethod("onTimeOut", new Class<?>[] { long.class });
	// onTimeOutMethod.setAccessible(true);
	// } catch (Exception e) {
	// throw new RuntimeException(e);
	// }
	// }

	/** 回调触发 **/
	@SuppressWarnings("unchecked")
	public void callback(TClusterSession session, final int callbackId, final int code, final String msg, final Object retObj) throws Exception {
		CallBackNode node = callbacks.remove(callbackId);
		if (node == null) {
			// System.out.println("no find callback or timeout! " + callbackId);
			return;
		}

		// action
		final CallBack<Object> callback = ((CallBack<Object>) node.callback);
		if (callback.isDirectCall()) {
			callback.onCallBack(code, msg, retObj); // 直接调用
			return;
		}

		// 提交任务
		this.serviceProxy.execute(new Runnable() {
			@Override
			public void run() {
				callback.onCallBack(code, msg, retObj); // 直接调用
			}
		});
		// onCallBackMethod.invoke(node.callback, code, msg, retObj);
	}

	/** 让所有回调超时移除 **/
	public void callBackAllTimeOut() {
		callBackTimeOut(callbacks.values());
	}

	/** 回调超时处理(删除回调并触发超时) **/
	protected void callBackTimeOut(Collection<CallBackNode> removes) {
		long nowTime = System.currentTimeMillis();
		// 执行删除
		for (CallBackNode node : removes) {
			if (!callbacks.remove(node.callbackId, node)) {
				continue; // 已经被别的移除了
			}
			// time out
			try {
				node.callback.onTimeOut(nowTime - node.startTime);
				// onTimeOutMethod.invoke(node.callback, nowTime - node.startTime);
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
		}
	}

	/** 回调节点 **/
	protected class CallBackNode {
		public int callbackId; // 回调Id
		public CallBack<?> callback; // 回调对象
		public long startTime; // 开始时间
	}

	/** 访问回调接口 **/
	public static abstract class CallBack<T> {
		/** 超时错误 **/
		public static int CODE_TIMEOUT = 0;
		/** 访问错误 **/
		public static int CODE_ERROR = -1;

		/** 回调触发 **/
		protected abstract void onCallBack(int code, String msg, T retObj);

		/** 超时处理 **/
		protected void onTimeOut(long time) {
			onCallBack(CODE_TIMEOUT, "timeout: " + time, null);
		}

		/** 超时时间 **/
		public int getMaxTimeOut() {
			return 30 * 1000;
		}

		/** 是否直接运行, 不提交到服务的线程池执行(不提交的情况会阻塞当前的IO线程, 不能运行复杂阻塞的逻辑) **/
		protected boolean isDirectCall() {
			return false;
		}
	}
}
